package com.hucais.sync.es2hive.service

import com.hucais.core.constant.Constants
import com.hucais.core.utils.{CommonUtils, DefaultPropertiesUtil}
import com.hucais.etl.common.service.CommonQueryService
import org.apache.spark.SparkContext
import org.apache.spark.sql.{Dataset, SaveMode, SparkSession}
import org.elasticsearch.spark.sql.EsSparkSQL

object SnycBookBaseInfoService {
  def action(ssc: SparkContext, sparkSession: SparkSession): Unit = {
    val hdfsPath: String = "/hucaisdata/published/dws/book_query_info/"

    val dataDS = SnycBookBaseInfoService.getEsData(sparkSession)
    dataDS.cache()
    CommonQueryService.saveAsFileAbsPath(dataDS.toDF(), hdfsPath, Constants.HIVE_SPLIT_STR, SaveMode.Overwrite)
  }

  private def getEsData(sparkSession: SparkSession): Dataset[EsData] = {
    import sparkSession.implicits._
    val esQuery =
      """
        |{
        |  "query": {
        |    "match_all": {}
        |  }
        |}
       """.stripMargin

    val tmpDS = EsSparkSQL.esDF(sparkSession, DefaultPropertiesUtil.get("dws.book.query.info"), esQuery).as[EsData]
    // 解决ES获取的数据和Hive对应字段顺序不一致
    tmpDS.mapPartitions(partitions => {
      partitions.map(item => {
        val store = CommonUtils.handleBlankStr(item.store)
        val store_comments = CommonUtils.handleBlankNumber(item.store_comments).toString
        val isbn = CommonUtils.handleBlankStr(item.isbn)
        val book_name = CommonUtils.handleBlankStr(item.book_name)
        val category = CommonUtils.handleBlankStr(item.category)
        val slogan = CommonUtils.handleBlankStr(item.slogan)
        val book_description = CommonUtils.handleBlankStr(item.book_description)
        val languages = CommonUtils.handleBlankStr(item.languages)
        val word_count = CommonUtils.handleBlankNumber(item.word_count).toString
        val book_comments = CommonUtils.handleBlankNumber(item.book_comments).toString
        val store_pricing = CommonUtils.handleBlankFloat(item.store_pricing).toString
        val selling_price = CommonUtils.handleBlankFloat(item.selling_price).toString
        val publishing_house = CommonUtils.handleBlankStr(item.publishing_house)
        val publishing_time = CommonUtils.handleBlankStr(item.publishing_time)
        val printing_time = CommonUtils.handleBlankStr(item.printing_time)
        val shelf_time = CommonUtils.handleBlankStr(item.shelf_time)
        val edition = CommonUtils.handleBlankStr(item.edition)
        val impression = CommonUtils.handleBlankStr(item.impression)
        val inventory = CommonUtils.handleBlankNumber(item.inventory).toString
        val sales = CommonUtils.handleBlankNumber(item.sales).toString
        val author = CommonUtils.handleBlankStr(item.sales)
        val shuppites = CommonUtils.handleBlankStr(item.shuppites)
        val format = CommonUtils.handleBlankStr(item.format)
        val is_suit = CommonUtils.handleBlankStr(item.is_suit)
        val suits = CommonUtils.handleBlankStr(item.suits)
        val binding_layout = CommonUtils.handleBlankStr(item.binding_layout)
        val pages = CommonUtils.handleBlankNumber(item.pages).toString
        val papers = CommonUtils.handleBlankStr(item.papers).toString
        val selling_stores = CommonUtils.handleBlankNumber(item.selling_stores).toString
        val published_year_range = CommonUtils.handleBlankStr(item.published_year_range)
        val published_year_integral = CommonUtils.handleBlankNumber(item.published_year_integral).toString
        val comments_range = CommonUtils.handleBlankStr(item.comments_range)
        val comments_integral = CommonUtils.handleBlankNumber(item.comments_integral).toString
        val premium_range = CommonUtils.handleBlankStr(item.premium_range)
        val premium_integral = CommonUtils.handleBlankNumber(item.premium_integral).toString
        val selling_stores_range = CommonUtils.handleBlankStr(item.selling_stores_range)
        val selling_stores_integral = CommonUtils.handleBlankNumber(item.selling_stores_integral).toString
        val total_integral = CommonUtils.handleBlankNumber(item.total_integral).toString
        val uploader = CommonUtils.handleBlankStr(item.uploader)
        val copyright_month = CommonUtils.handleBlankStr(item.copyright_month)

        EsData(
          item.first_channel, item.second_channel, store, store_comments, isbn,
          book_name, category, slogan, book_description, languages,
          word_count, book_comments, store_pricing, selling_price,
          publishing_house, publishing_time, printing_time, shelf_time,edition,
          impression,inventory, sales, author, shuppites, format,
          is_suit, suits, binding_layout, pages, papers,
          selling_stores, published_year_range, published_year_integral, comments_range,
          comments_integral, premium_range, premium_integral, selling_stores_range,
          selling_stores_integral, total_integral, uploader, copyright_month, item.create_type, item.create_time, item.update_time
        )
      })
    })
  }
}

case class EsData(
                   first_channel: String,
                   second_channel: String,
                   store: String,
                   store_comments: String,
                   isbn: String,
                   book_name: String,
                   category: String,
                   slogan: String,
                   book_description: String,
                   languages: String,
                   word_count: String,
                   book_comments: String,
                   store_pricing: String,
                   selling_price: String,
                   publishing_house: String,
                   publishing_time: String,
                   printing_time: String,
                   shelf_time: String,
                   edition: String,
                   impression: String,
                   inventory: String,
                   sales: String,
                   author: String,
                   shuppites: String,
                   format: String,
                   is_suit: String,
                   suits: String,
                   binding_layout: String,
                   pages: String,
                   papers: String,
                   selling_stores: String,
                   published_year_range: String,
                   published_year_integral: String,
                   comments_range: String,
                   comments_integral: String,
                   premium_range: String,
                   premium_integral: String,
                   selling_stores_range: String,
                   selling_stores_integral: String,
                   total_integral: String,
                   uploader: String,
                   copyright_month: String,
                   create_type: String,
                   create_time: String,
                   update_time: String
                 )
